package com.fanxuankai.kafka.connect.sink.redis;

import com.fanxuankai.kafka.connect.sink.redis.config.RedisConnectorConfig;
import com.fanxuankai.kafka.connect.sink.redis.config.RedisSinkConnectorConfig;
import com.fanxuankai.kafka.connect.sink.redis.consumer.JsonSinkRecordConsumer;
import com.fanxuankai.kafka.connect.sink.redis.consumer.SinkRecordConsumer;
import com.fanxuankai.kafka.connect.sink.redis.consumer.SpringSinkRecordConsumer;
import io.lettuce.core.ClientOptions;
import io.lettuce.core.SocketOptions;
import io.lettuce.core.SslOptions;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;
import org.springframework.data.redis.connection.*;
import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.util.Collection;
import java.util.HashSet;
import java.util.Map;

/**
 * @author fanxuankai
 */
public class RedisSinkTask extends SinkTask {
    private RedisTemplate<String, Object> redisTemplate;
    private SinkRecordConsumer consumer;

    @Override
    public String version() {
        return "0.0.1";
    }

    @Override
    public void start(Map<String, String> props) {
        RedisSinkConnectorConfig config = new RedisSinkConnectorConfig(props);
        redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory(config));
        redisTemplate.setKeySerializer(RedisSerializer.string());
        redisTemplate.setValueSerializer(RedisSerializer.string());
        redisTemplate.setHashKeySerializer(RedisSerializer.string());
        redisTemplate.setHashValueSerializer(RedisSerializer.string());
        redisTemplate.afterPropertiesSet();
        if (config.storageFormat == RedisSinkConnectorConfig.StorageFormat.Json) {
            consumer = new JsonSinkRecordConsumer(redisTemplate);
        } else if (config.storageFormat == RedisSinkConnectorConfig.StorageFormat.Spring) {
            consumer = new SpringSinkRecordConsumer(redisTemplate);
        }
    }

    @Override
    public void put(Collection<SinkRecord> records) {
        consumer.accept(records);
    }

    @Override
    public void stop() {
        redisTemplate = null;
        consumer = null;
    }

    private RedisConnectionFactory redisConnectionFactory(RedisSinkConnectorConfig config) {
        final SocketOptions socketOptions = SocketOptions.builder()
                .tcpNoDelay(config.tcpNoDelay)
                .connectTimeout(Duration.ofMillis(config.connectTimeout))
                .keepAlive(config.keepAliveEnabled)
                .build();
        final ClientOptions.Builder clientOptions = ClientOptions.builder()
                .socketOptions(socketOptions)
                .requestQueueSize(config.requestQueueSize)
                .autoReconnect(config.autoReconnectEnabled);
        if (config.sslEnabled) {
            SslOptions.Builder builder = SslOptions.builder();
            switch (config.sslProvider) {
                case JDK:
                    builder.jdkSslProvider();
                    break;
                case OPENSSL:
                    builder.openSslProvider();
                    break;
                default:
                    throw new UnsupportedOperationException(String.format("%s is not a supported value for %s.",
                            config.sslProvider, RedisConnectorConfig.SSL_PROVIDER_CONFIG));
            }
            if (null != config.keystorePath) {
                if (null != config.keystorePassword) {
                    builder.keystore(config.keystorePath, config.keystorePassword.toCharArray());
                } else {
                    builder.keystore(config.keystorePath);
                }
            }
            if (null != config.truststorePath) {
                if (null != config.truststorePassword) {
                    builder.truststore(config.truststorePath, config.keystorePassword);
                } else {
                    builder.truststore(config.truststorePath);
                }
            }
            clientOptions.sslOptions(builder.build());
        }
        RedisConfiguration redisConfiguration;
        LettuceClientConfiguration clientConfiguration = LettuceClientConfiguration.builder()
                .commandTimeout(Duration.ofMillis(config.operationTimeoutMs))
                .clientOptions(clientOptions.build())
                .build();
        if (RedisConnectorConfig.ClientMode.Cluster == config.clientMode) {
            RedisClusterConfiguration clusterConfiguration = new RedisClusterConfiguration(config.hosts);
            if (StringUtils.hasText(config.password)) {
                clusterConfiguration.setPassword(config.password);
            }
            redisConfiguration = clusterConfiguration;
        } else if (RedisConnectorConfig.ClientMode.Standalone == config.clientMode) {
            String[] split = config.hosts.get(0)
                    .split(":");
            RedisStandaloneConfiguration standaloneConfiguration = new RedisStandaloneConfiguration(split[0],
                    Integer.parseInt(split[1]));
            standaloneConfiguration.setDatabase(config.database);
            if (StringUtils.hasText(config.password)) {
                standaloneConfiguration.setPassword(config.password);
            }
            redisConfiguration = standaloneConfiguration;
        } else if (RedisConnectorConfig.ClientMode.Sentinel == config.clientMode) {
            RedisSentinelConfiguration sentinelConfiguration = new RedisSentinelConfiguration(config.sentinelMaster,
                    new HashSet<>(config.hosts));
            sentinelConfiguration.setDatabase(config.database);
            if (StringUtils.hasText(config.password)) {
                sentinelConfiguration.setPassword(config.password);
            }
            redisConfiguration = sentinelConfiguration;
        } else {
            throw new UnsupportedOperationException(String.format("%s is not supported", config.clientMode));
        }
        LettuceConnectionFactory factory = new LettuceConnectionFactory(redisConfiguration, clientConfiguration);
        factory.afterPropertiesSet();
        return factory;
    }
}
